package com.yifeng.repo.toolkit.mq;

import com.google.common.base.Strings;
import com.yifeng.repo.toolkit.mq.configure.MqProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;

import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;

/**
 * Created by daibing on 2022/9/28.
 */
@Slf4j
public class TempMessageConsumer implements Consumer<Message<String>> {
    private final MqProperties properties;

    public TempMessageConsumer(MqProperties properties) {
        this.properties = properties;
    }

    @Override
    public void accept(Message<String> msg) {
        // extract base data
        String tag = this.getHeader(msg.getHeaders(), "ROCKET_TAGS", "RABBIT_TAGS");
        String key = this.getHeader(msg.getHeaders(), "ROCKET_KEYS", "RABBIT_KEYS");
        String body = msg.getPayload();
        Map<String, Object> headers = getHeaders(msg.getHeaders());

        // print data
        log.info("{} Consumer tag={} key={} Receive New Messages run: body={}, headers={}, routingKey={}",
                Thread.currentThread().getName(), tag, key, body, headers, msg.getHeaders().get("x-routing-key"));
    }

    private String getHeader(MessageHeaders headers, String key1, String key2) {
        String header = headers.get(key1, String.class);
        if (!Strings.isNullOrEmpty(header)) {
            return header;
        }
        return headers.get(key2, String.class);
    }

    private Map<String, Object> getHeaders(MessageHeaders headers) {
        Map<String, Object> map = new HashMap<>();
        for (Map.Entry<String, Object> entry : headers.entrySet()) {
            if (entry.getKey().startsWith(properties.getBizHeaderPrefix())) {
                String key = entry.getKey().substring(properties.getBizHeaderPrefix().length());
                map.put(key, entry.getValue());
            }
        }
        return map;
    }
}
